package com.whoami.dataplatform.tag_combine;

import org.apache.commons.lang3.StringUtils;
import org.apache.commons.lang3.math.NumberUtils;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.conf.Configured;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.NullWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Job;
import org.apache.hadoop.mapreduce.Mapper;
import org.apache.hadoop.mapreduce.lib.input.FileInputFormat;
import org.apache.hadoop.mapreduce.lib.input.SequenceFileAsBinaryInputFormat;
import org.apache.hadoop.mapreduce.lib.output.FileOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.NullOutputFormat;
import org.apache.hadoop.mapreduce.lib.output.TextOutputFormat;
import org.apache.hadoop.util.Tool;
import org.apache.hadoop.util.ToolRunner;

import java.io.FileInputStream;
import java.io.IOException;
import java.nio.ByteBuffer;
import java.util.BitSet;
import java.util.Properties;
import java.util.Stack;

/**
 * 接上一个任务输出的 tid_bits -> uid
 * 24 209 & 24 208 & |
 * 转为tid_bits -> 3, uid \n uid \n uid
 *
 * @author tzp
 * @since 2019/7/31
 */
@SuppressWarnings("Duplicates")
public class TidbitsCombineFilter extends Configured implements Tool {
    /**
     * 给MR框架的一堆配置, 文件里面的内容会被set到mr conf里
     */
    public static final String PARAMS_MR_PROP = "mr_properties";

    @Override
    public int run(String[] args) throws Exception {
        Configuration conf = getConf();
        String mrProFile = conf.get(PARAMS_MR_PROP);
        if (StringUtils.isNotEmpty(mrProFile)) {
            Properties mrProperties = new Properties();
            mrProperties.load(new FileInputStream(mrProFile));
            for (Object key : mrProperties.keySet()) {
                conf.set(key.toString(), mrProperties.get(key).toString());
            }
        }
        String date = args[0];//20190723
        String input = args[1];
        String output = args[2];//outputbase
        String exps = args[3];//表达式
        Boolean needOutput = Boolean.valueOf(args[4]);//是否输出id
        conf.set("expression", exps);
        conf.setBoolean("needOutput", needOutput);

        Job job = Job.getInstance(conf, this.getClass().getSimpleName() + " " + date);
        job.setJarByClass(this.getClass());

        //输入
        job.setInputFormatClass(SequenceFileAsBinaryInputFormat.class);
        FileInputFormat.addInputPath(job, new Path(input));

        //mapper
        job.setMapperClass(FilterMapper.class);
        job.setMapOutputKeyClass(BytesWritable.class);
        job.setMapOutputValueClass(BytesWritable.class);

        //reducer
        job.setNumReduceTasks(0);

        //输出
        if (needOutput) {
            job.setOutputFormatClass(TextOutputFormat.class);
            FileOutputFormat.setOutputPath(job, new Path(output));
        } else {
            job.setOutputFormatClass(NullOutputFormat.class);
        }

        boolean result = job.waitForCompletion(true);

        return result ? 0 : 1;
    }

    public static class FilterMapper extends Mapper<BytesWritable, BytesWritable, NullWritable, Text> {
        Text outValue = new Text();
        String[] exps;
        boolean needOutput;

        @Override
        protected void setup(Context context) throws IOException, InterruptedException {

            Configuration conf = context.getConfiguration();
            exps = conf.get("expression").split(" ");
            needOutput = conf.getBoolean("needOutput", false);
        }

        @Override
        protected void map(BytesWritable key, BytesWritable value, Context context) throws IOException, InterruptedException {
            BitSet bitSet = BitSet.valueOf(key.getBytes());//多点无效的没关系
            if (eval(bitSet, exps)) {
                ByteBuffer byteBuffer = ByteBuffer.wrap(value.getBytes());
                int num = byteBuffer.getInt();
                if (needOutput) {
                    outValue.set(value.getBytes(), 4, value.getLength() - 4);
                    context.write(NullWritable.get(), outValue);
                }
                context.getCounter("map filter", "pass uids").increment(num);
                context.getCounter("map filter", "pass line").increment(1);
            }
            context.getCounter("map filter", "total line").increment(1);
        }

        /**
         * 执行布尔后缀表达式: "24 209 & 24 208 & |"
         */
        public static boolean eval(BitSet bitSet, String[] exps) {
            Stack<Boolean> temp = new Stack<>();
            for (String exp : exps) {
                if (NumberUtils.isNumber(exp)) {
                    temp.push(bitSet.get(Integer.parseInt(exp)));
                } else if ("&".equals(exp)) {
                    Boolean operand1 = temp.pop();
                    Boolean operand2 = temp.pop();
                    temp.push(operand1 & operand2);
                } else if ("|".equals(exp)) {
                    Boolean operand1 = temp.pop();
                    Boolean operand2 = temp.pop();
                    temp.push(operand1 | operand2);
                } else if ("^".equals(exp)) {
                    Boolean operand1 = temp.pop();
                    temp.push(!operand1);
                }
            }
            Boolean result = temp.pop();
            assert temp.size() == 0;
            return result;
        }
    }

    public static void main(String[] args) throws Exception {
        System.exit(ToolRunner.run(new Configuration(), new TidbitsCombineFilter(), args));
    }
}
